package com.ds.infrastructure.hyperspace.container.internal.opentracing.reporter;

import com.ds.infrastructure.hyperspace.common.config.OpentracingConfig;
import com.ds.infrastructure.hyperspace.container.internal.HyperspaceThreadFactory;
import com.ds.infrastructure.hyperspace.container.internal.NettyContainer;
import com.ds.infrastructure.hyperspace.container.internal.opentracing.impl.TracerSpan;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import lombok.extern.slf4j.Slf4j;
import org.jctools.queues.MpscArrayQueue;

import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.*;

/**
 * @author yaozh
 */
@Slf4j
public class OpentracingReporter implements IOpentracingReporter {
    private NettyContainer nettyContainer;
    private ExecutorService monitorPool = new ThreadPoolExecutor(1, 1,
            0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<Runnable>(),
            new HyperspaceThreadFactory("OpentracingMonitor"));
    private OpentracingConfig opentracingConfig;
    private OpentracingSender opentracingSender;
    private static final MpscArrayQueue<TracerSpan> SPAN_DATA_QUEUE = new MpscArrayQueue<>(100000);
    private static final CountDownLatch EMPTY_SPAN_DATA_QUEUE_LATCH = new CountDownLatch(1);
    private static final int EMPTY_SPAN_SLEEP_TIME_MILL = 200;
    private static final int EMPTY_QUEUE_SLEEP_TIME_MILL = 1000;
    private int maxSendAwaitTimeMill;
    private Charset charset = StandardCharsets.UTF_8;

    public OpentracingReporter(NettyContainer container) {
        nettyContainer = container;
        opentracingConfig = nettyContainer.getOpentracingConfig();
        maxSendAwaitTimeMill = getMaxSendAwaitSeconds(opentracingConfig.getMaxSendAwaitSeconds());
        opentracingSender = new OpentracingSender(nettyContainer.getOpentracingReporterEventLoopGroup(), getSenderUrl());
        monitorPool.submit(this::report);
    }

    @Override
    public void reload(OpentracingConfig opentracingConfig) {
        this.opentracingConfig = opentracingConfig;
        this.maxSendAwaitTimeMill = getMaxSendAwaitSeconds(this.opentracingConfig.getMaxSendAwaitSeconds());
        this.opentracingSender.reload(getSenderUrl());
    }

    @Override
    public void save(TracerSpan span) {
        if (!opentracingConfig.isEnable() || !opentracingSender.isEnable() || span == null) {
            return;
        }
        if (SPAN_DATA_QUEUE.offer(span)) {
            return;
        }
        nettyContainer.getMetricsManager().incrementTracerDiscardedSpanCounter();
    }


    private void report() {
        while (true) {
            if (SPAN_DATA_QUEUE.isEmpty()) {
                sleep(EMPTY_QUEUE_SLEEP_TIME_MILL);
                continue;
            }
            ByteBuf buffer = Unpooled.buffer(getSpanDataInitCapacity());
            boolean autoRelease = false;
            try {
                if (fillData(buffer)) {
                    autoRelease = opentracingSender.send(buffer);
                }
            } catch (Exception e) {
                log.error("Opentracing reporter send data exception!", e);
            } finally {
                if (!autoRelease && buffer.refCnt() > 0) {
                    buffer.release();
                }
            }
        }
    }

    private void sleep(long timeoutMill) {
        try {
            EMPTY_SPAN_DATA_QUEUE_LATCH.await(timeoutMill, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            log.error("Opentracing reporter sleep interrupted exception!", e);
        }
    }

    private boolean fillData(ByteBuf buffer) {
        buffer.writeCharSequence("[", charset);
        int sleepTotalTime = 0;
        int i = opentracingConfig.getMaxBatchSendSize();
        while (i > 0) {
            TracerSpan span = SPAN_DATA_QUEUE.poll();
            if (span != null) {
                if (i < opentracingConfig.getMaxBatchSendSize()) {
                    buffer.writeCharSequence(",", charset);
                }
                i--;
                spanToString(buffer, span);
                continue;
            }
            if (sleepTotalTime >= maxSendAwaitTimeMill) {
                break;
            }
            sleepTotalTime += EMPTY_SPAN_SLEEP_TIME_MILL;
            sleep(EMPTY_SPAN_SLEEP_TIME_MILL);
        }
        if (i == opentracingConfig.getMaxBatchSendSize()) {
            return false;
        }
        buffer.writeCharSequence("]", charset);
        return true;
    }

    private void spanToString(ByteBuf buffer, TracerSpan span) {

        buffer.writeCharSequence("{\"traceId\":\"", charset);
        buffer.writeCharSequence(span.getTraceId(), charset);
        buffer.writeCharSequence("\",\"spanId\":\"", charset);
        buffer.writeCharSequence(span.getSpanId(), charset);
        buffer.writeCharSequence("\",\"parentSpanId\":\"", charset);
        buffer.writeCharSequence(span.getParentId(), charset);
        buffer.writeCharSequence("\",\"startTime\":", charset);
        buffer.writeCharSequence(String.valueOf(span.getStartMillis()), charset);
        buffer.writeCharSequence(",\"finishTime\":", charset);
        buffer.writeCharSequence(String.valueOf(span.getFinishMillis()), charset);
        buffer.writeCharSequence(",\"operationName\":\"", charset);
        buffer.writeCharSequence(span.getOperationName(), charset);
        buffer.writeCharSequence("\",\"tags\":{", charset);
        int i = 0;
        Iterator<Map.Entry<String, Object>> entries = span.getTags().entrySet().iterator();
        while (entries.hasNext()) {
            if (i++ > 0) {
                buffer.writeCharSequence(",", charset);
            }
            Map.Entry<String, Object> entry = entries.next();
            if (entry.getValue() instanceof Boolean) {
                buffer.writeCharSequence("\"", charset);
                buffer.writeCharSequence(entry.getKey(), charset);
                buffer.writeCharSequence("\":", charset);
                buffer.writeCharSequence(String.valueOf(entry.getValue()), charset);
            } else {
                buffer.writeCharSequence("\"", charset);
                buffer.writeCharSequence(entry.getKey(), charset);
                buffer.writeCharSequence("\":\"", charset);
                buffer.writeCharSequence(String.valueOf(entry.getValue()), charset);
                buffer.writeCharSequence("\"", charset);
            }
        }
        buffer.writeCharSequence("}}", charset);
    }

    private String getSenderUrl() {
        return opentracingConfig.isEnable() ? opentracingConfig.getUrl() : null;
    }

    private int getMaxSendAwaitSeconds(int maxSendAwaitSeconds) {
        return maxSendAwaitSeconds * 1000;
    }

    /**
     * //1292*2000
     *
     * @return
     */
    private int getSpanDataInitCapacity() {
        int initSize = Math.min(opentracingConfig.getMaxBatchSendSize(), SPAN_DATA_QUEUE.size() + 1);
        return initSize * 1500;
    }

    @Override
    public void shutdown() {
        monitorPool.shutdown();
        opentracingSender.shutdown();
        log.info("Shutdown opentracing reporter success!");
    }
}
